package com.djzhu.lang.threadpool;

import java.util.concurrent.*;

/**
 * 演示了线程池相关的逻辑代码
 *
 * @author DJ.Zhu
 */
public class ThreadPoolDemo {

    public static void main(String[] args) throws InterruptedException, ExecutionException {
//        testSingleThreadExecutor();
//        testFixedThreadExecutor();
//        testExecutorService();
//        testThreadPoolExecutor();
//        testScheduleThreadExecutor();
    }

    /**
     * 使用单个线程，队列无限。线程池保证任意时间点只有一个任务在执行，且可以容忍工作线程的失败（失败后会新建线程来继续工作）。
     * corePoolSize:1
     * maximumPoolSize:1;
     * keepAlive:0    //即永远保持
     * workQueue:LinkedBlockingQueue
     * @throws InterruptedException
     */
    public static void testSingleThreadExecutor() throws InterruptedException {
        Executor executor = Executors.newSingleThreadExecutor();
        executor.execute(new Runnable() {
            @Override
            public void run() {
                System.out.println("hello world");
                System.out.println(Thread.currentThread().getId());
                throw new RuntimeException();
            }
        });
        Thread.sleep(1000); //等待一秒钟，让上面的任务跑完
        executor.execute(new Runnable() {
            @Override
            public void run() {
                System.out.println("hello world");
                System.out.println(Thread.currentThread().getId());
                throw new RuntimeException();
            }
        });
    }

    /**
     * 永远保证任意时间点都有nThreads个线程，如果线程失败，创建新的线程来代替他。
     * corePoolSize:nThreads
     * maximumPoolSize:nThreads;
     * keepAlive:0    //即永远保持
     * workQueue:LinkedBlockingQueue
     * @throws InterruptedException
     */
    public static void testFixedThreadExecutor() throws InterruptedException {
        Executor executor = Executors.newFixedThreadPool(1);
        for (int i = 0; i < 10; i++){
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    System.out.println("hello world");
                    System.out.println(Thread.currentThread().getId());
                    throw new RuntimeException();
                }
            });
            Thread.sleep(1000); //等待一秒钟，让上面的任务跑完
        }
    }

    /**
     * ExecutorService继承自Executor，比Executor多支持一些功能。可以submit Runnable & Callable以获取Future。
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public static void testExecutorService() throws ExecutionException, InterruptedException {
        ExecutorService executorService = Executors.newSingleThreadExecutor();
        Future future = executorService.submit(new Runnable() {
            @Override
            public void run() {
                System.out.println("submit with Runnable.");
            }
        });
        System.out.println("return with Runnable: " + future.get());
        future = executorService.submit(new Callable<Object>() {
            @Override
            public Object call() throws Exception {
                System.out.println("submit with Callable.");
                return 1;
            }
        });
        System.out.println("return with Callable: " + future.get());
        future = executorService.submit(new Runnable() {
            @Override
            public void run() {
                System.out.println("submit with Callable.");
            }
        }, 23);//return 23 if success
        System.out.println("return with Runnable and result: " + future.get());
    }

    /**
     * 跟其余常见的线程池不同，它继承了ThreadPoolExecutor(其余线程池只是ThreadPoolExecutor的不同配置)，
     * 并实现一个interface ScheduledExecutorService extends ExecutorService
     * corePoolSize:nThreads
     * maximumPoolSize:Integer.MAX_VALUE;
     * keepAlive:60s
     * workQueue:DelayedWorkQueue
     */
    public static void testScheduleThreadExecutor(){
        ScheduledExecutorService threadPoolExecutor = Executors.newScheduledThreadPool(1);
        threadPoolExecutor.schedule(new Runnable() {
            @Override
            public void run() {
                System.out.println("Current thread:" + Thread.currentThread().getId());
            }
        }, 3, TimeUnit.SECONDS);
        threadPoolExecutor.scheduleWithFixedDelay(new Runnable() {
            @Override
            public void run() {
                System.out.println("current thread:" + Thread.currentThread().getId());
                throw new RuntimeException();
            }
        }, 0,3000, TimeUnit.MILLISECONDS);
        threadPoolExecutor.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                System.out.println("current thread:" + Thread.currentThread().getId());
                throw new RuntimeException();
            }
        }, 0,3000, TimeUnit.MILLISECONDS);
    }

    /**
     * 本质上上述所有的线程池都是对ThreadPoolExecutor的不同配置
     * @throws InterruptedException
     */
    public static void testThreadPoolExecutor() throws InterruptedException {
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(3, 5, 10,
                TimeUnit.SECONDS, new LinkedBlockingQueue<>());
        System.out.println("begin active count: " + threadPoolExecutor.getActiveCount());
        new Thread(new Runnable() {
            @Override
            public void run() {
                while (true){
                    int activeCnt = threadPoolExecutor.getActiveCount();
                    long completeTask = threadPoolExecutor.getCompletedTaskCount();
                    long taskCnt = threadPoolExecutor.getTaskCount();
                    System.out.println("current active count:" + activeCnt + ", completeTask:" + completeTask + ", taskCnt:" + taskCnt);
                    try {
                        Thread.sleep(500);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }).start();
        for (int i = 0; i<10; i++){
            threadPoolExecutor.submit(new Runnable() {
                @Override
                public void run() {
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });
            Thread.sleep(200);
        }

    }

}
